package day0613.managersystem

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration._
import scala.collection.mutable
import scala.concurrent.ExecutionContext.Implicits.global

/**
  * @program: hadoop
  * @description: 从节点
  * @author: Qiang.Ye
  * @date: 2019-06-15 16:42
  **/
class Worker extends Actor{

  //Worker给Master发送信号，需要Master引用
  var master:ActorSelection = null

  //生成UUID，作为Worker标识
  val id = UUID.randomUUID().toString

  override def preStart(): Unit = {
    //执行完构造方法后，执行的一次次方法

    //Worker向system发送链接请求
    master = context.system.actorSelection("akka.tcp://MasterActorSystem@127.0.0.1:8888/user/Master")
    //Worker向master发送注册信息
    master ! RegisterWorker(id,"127.0.0.1","10240","8")

  }

  override def receive: Receive = {

    case  RegisteredWorker(masterUrl)=>{
      //启动定时任务，发送心跳
      context.system.scheduler.schedule(0 millis,5000 millis ,self,HeartBeat)
    }
    case HeartBeat => {
      println("sent hearbeat")
      master ! SendHeartBeat(id)
    }
  }
}

object Worker{
  def main(args: Array[String]): Unit = {
    val clientPort = 8892

    //创建ActorSystem
    val configStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.port = $clientPort
       """.stripMargin
    val config = ConfigFactory.parseString(configStr)
    val actorSystem = ActorSystem("WorkerActorSystem",config)

    actorSystem.actorOf(Props[Worker],"Worker")

  }
}

